/* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements.  See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License.  You may obtain a copy of the License at
*
*     http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/


/*************************************************************************
 * File: controller.cpp
 *************************************************************************
 *\author: starlove
 *************************************************************************/

#include <network/epoll/controller.h>
#include <network/network.h>
#include <errno.h>

using namespace SmartEpoll;

CShareQueue<LPEVENT> CController::m_evProcessed;

CController::CController()
    :m_op(0)
{
    memset(m_ptIO, 0, sizeof(pthread_t) * __MLSE_WORK_THREAD_NUM__);
}

CController::~CController()
{
    Stop();
}

void CController::SetIOHandle(HANDLE op)
{
    Stop();
    m_op = op;
}

void CController::Start()
{
    m_op = !m_op ? epoll_create(__MLSE_EVENT_LIMIT__) : m_op;
    for(int i = 0; i < __MLSE_WORK_THREAD_NUM__; i ++)
    {
        pthread_create(&m_ptIO[i], NULL, CController::IOProcessFunc, this);
    }
}

void CController::Stop()
{
    m_op = -1;
    for(int i = 0; i < __MLSE_WORK_THREAD_NUM__; i ++)
        if(m_ptIO[i] > 0)
            pthread_join(m_ptIO[i], NULL);
    memset(m_ptIO, 0, sizeof(pthread_t) * __MLSE_WORK_THREAD_NUM__);
}

HANDLE CController::GetIOHandle()
{
    return m_op;
}

LPEVENT CController::PopEvent()
{
    return m_evProcessed.pop();
}

void *CController::IOProcessFunc(void *p)
{
    char szTmpData[__MLSE_EPOLL_RECV_BUFFSIZE__];
    memset(szTmpData, 0, __MLSE_EPOLL_RECV_BUFFSIZE__);
    epoll_event events[__MLSE_WORK_THREAD_EVENT_NUM__];
    CController *pController = (CController *)p;
    while(pController->GetIOHandle() > 0)
    {
        memset(events, 0, sizeof(epoll_event) * __MLSE_WORK_THREAD_EVENT_NUM__);
        int nRet = epoll_wait(
                       pController->GetIOHandle(),
                       events,
                       __MLSE_WORK_THREAD_EVENT_NUM__,
                       1
                   );

		//if(nRet > 0)
		//	DEBUG_INFO_NETWORK(0, "Have %d events, errno: %d", nRet, errno);
        for(int i = 0; i < nRet; i ++)
        {
            epoll_event *pevt = &events[i];
            LPEVENT ev = (LPEVENT)pevt->data.ptr;
				if (!ev)
					continue;

            SOCKET sck = (SOCKET)ev->hd;

            ev->event_sets = ETP_UNKNOWN;
            if (MASK_TEST(ev->events, ETP_NEW))
            {
                sockaddr_in addr;
                socklen_t ts = sizeof(sockaddr_in);
                SOCKET s = ::accept(sck, (sockaddr *)&addr, &ts);
                if (s == INVALID_SOCKET)
                {
                    continue;
                }
                ev->hd = (EVHANDLE)s;
                ev->event_sets |= ETP_NEW;
            }
            else
            {
                IPeer *pPeer = (IPeer *)ev->data;
                if (MASK_TEST(pevt->events, EPOLLIN))
                {
                    LPNETWORKBUFFER buf = pPeer->GetRecvBuffer();
                    ev->datasize = 0;
                    int re = 0;
                    int wait_size = __MLSE_EPOLL_RECV_BUFFSIZE__;
//					if (buf->GetDataSize() > sizeof(PACKET_FIXED_HEADER))
//					{
//						wait_size = (buf->GetSize() - buf->GetDataSize());
//					}
                    if ( (re = ::recv(sck, szTmpData, std::min<int>(wait_size, __MLSE_EPOLL_RECV_BUFFSIZE__), 0)) > 0 )
                    {
                        buf->Append(szTmpData, re);
                        ev->datasize += re;
                    }
                    if ( 0 < re || 0 == errno || EAGAIN == errno || EWOULDBLOCK == errno )
                    {
                        DEBUG_INFO_NETWORK(sck, "Recv length: %d", ev->datasize);
                        ev->event_sets |= ETP_READ_READY;
                    }
                    else
                    {
                        DEBUG_INFO_NETWORK(sck, "Recv errno: %d", errno);
                        ev->event_sets |= ETP_IOERROR;
                    }
                }
                if (MASK_TEST(pevt->events, EPOLLOUT))
                {
					ev->event_sets |= ETP_WRITE_READY;
                }
            }
            if (ETP_UNKNOWN == ev->event_sets || MASK_TEST(pevt->events, EPOLLHUP))
            {
                //DEBUG_INFO_NETWORK(sck, "IO ERROR");
                ev->event_sets |= ETP_IOERROR;
            }
            m_evProcessed.push(ev);
        }
    }
    return NULL;
}
